/* Copyright (C) 2008 Sahid Ferdjaoui <sahid@funraill.org>
 *
 *   This file is part of Postmemd.
 *
 * Postmemd is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */


#include "pm_poll.h"
#include "pm_log.h"
#include "pm_core.h"
#include "pm_config.h"

#include <ev.h>
#include <json.h>

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>

extern io_poller io;
extern config conf;


int
pm_socket_create (sfd_t * sock, struct in_addr interface, int port, int backlog)
{
	struct sockaddr_in addr;

	if (( * sock = socket (AF_INET, SOCK_STREAM, 0)) < 0)
		{
			pm_log_error ("pm_socket_create () - socket ()", 1);
			return -1;
		}

	memset (& addr, 0, sizeof (struct sockaddr_in));
	addr.sin_family = AF_INET;
	addr.sin_port = port;
	addr.sin_addr = interface;
	
	if (bind ( * sock, (struct sockaddr * ) & addr, sizeof (addr)) < 0)
		{
			close ( * sock);
			pm_log_error ("pm_socket_create () - bind ()", 1);
			return -1;
		}
	
	if (listen ( * sock, backlog) < 0)
		{
			close ( * sock);
			pm_log_error ("pm_socket_create () - listen ()", PM_PRINT_ERRNO);
			return -1;
		}

	return 0;
}

void
pm_record_fd (EV_P_ struct ev_io * event, int fd, void * handler, int revent)
{
	fprintf (stdout, "record : %d\n", fd);
	ev_io_init (event, handler, fd, revent);
	ev_io_start (EV_A_ event);
}

void
pm_remove_fd (EV_P_ struct ev_io * event)
{
	ev_io_stop (EV_A_ event);
}

void 
pm_event_handler (EV_P_ struct ev_io * w, int revents)
{
	socklen_t							sfdsize;
	sfd_t									client;
	core_t							*	core;
  struct sockaddr_in    addr;

	fprintf (stdout, "!!! NEW EVENT ON %d !!! \n", w->fd);

	/* new client */
	if (w->fd == io.listener)
		{
			if ((client = accept (io.listener, (struct sockaddr * ) & addr, & sfdsize)) < 0)
				{
					pm_log_error ("pm_event_handler () - accept ()", PM_PRINT_ERRNO);
					close (client);
					return;
				}
			
			fprintf (stdout, "nouveaux client %d\n", client);

			/** initialize new core for new connection */
			if (pm_core_init (EV_A_ client, pm_read_client, & core) == -1)
				{
					pm_log_error ("pm_event_handler () - pm_core_init ()", PM_PRINT_ERRNO);
					close (client);
					return;
				}


			/** record core in event loop */
			pm_record_fd (EV_A_ & core->event, core->sfd, pm_event_handler, EV_READ);
			pm_record_fd (EV_A_ & core->event, core->slave->read, pm_event_handler, EV_READ);
		}
	else 
		{
			core = (core_t * ) w->data;
			if (!core)
				{
					pm_log_error ("pm_event_handler () - sanity... no core in w->data", PM_NOPRINT_ERRNO);
					return;
				}
			fprintf (stdout, "event recu sur %d, etat : %d\n", w->fd, core->state);
			pm_core_process_loop (core);
		}
}

void
pm_event_handler_slaves (EV_P_ struct ev_io * w, int revents)
{
	slave_t * slave;


	fprintf (stdout, "!!! SLAVE NEW EVENT ON %d !!! \n", w->fd);

	slave = (slave_t * ) w->data;
	if (slave->read != w->fd)
		{
			pm_log_error ("pm_event_handler_slaves () - sanity !", PM_NOPRINT_ERRNO);
			return;
		}
	fprintf (stdout, "le slave a recu un signal\n");
	pm_slaves_process_loop (slave);
}

int
pm_poll_read_fd (int fd, char ** buffer, int * buffer_bytes)
{
	int rbytes = 0;
	fprintf (stdout, "read on %d, data: %s, bytes: %d\n", fd, * buffer, * buffer_bytes);
	/** initialize read buffer */
	if (* buffer == NULL)
		{
			* buffer = malloc (conf.max_buffer_size);
			if (* buffer == NULL)
				{
					pm_log_error ("pm_poll_read_fd () - malloc ()", PM_PRINT_ERRNO);
					sleep (1);
					return -1;
				}
			memset (* buffer, '\0', conf.max_buffer_size);
			* buffer_bytes = 0;
		}

	/** read file descriptor */
	rbytes = read (fd, * buffer + * buffer_bytes, conf.max_buffer_size - * buffer_bytes);
	if (rbytes > 0)
		{
			* buffer_bytes += rbytes;
			return 1;
		}
	if ((rbytes == 0) ||	((rbytes == -1) && (errno == EAGAIN)))
		{
			return 0;
		}
	pm_log_error ("pm_poll_read_fd () - read ()", PM_PRINT_ERRNO);
	return -1;
}

int
pm_poll_write_fd (int fd, char * buffer, int * buffer_bytes, int * writed)
{
	int wbytes = 0;
	fprintf (stdout, "write on %d, data: %s, bytes: %d, writed: %d\n", fd, buffer, * buffer_bytes, * writed);
	wbytes = write (fd, buffer + * writed, * buffer_bytes - * writed);
	if (wbytes > 0)
		{
			* writed += wbytes;
			return 1;
		}
	if ((wbytes == 0) || ((wbytes == -1) && (errno == EAGAIN)))
		{
			return 0;
		}
	pm_log_error ("pm_poll_wirte_fd () - write ()", PM_PRINT_ERRNO);
	return -1;
}

